package zdz.stormtopo;


import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import zdz.util.DatabaseUtil;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by Simon on 2018/10/23.
 */
public class CountBolt extends BaseRichBolt {

    private int count,offset;

    Connection conn=null;
    PreparedStatement ps = null;
    ResultSet rs = null;

    private Map<String,Integer>wordList;

    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        wordList = new HashMap<>();
        conn= DatabaseUtil.getConnection();
        count=0;
        offset=10;
    }

    @Override
    public void cleanup() {
        try {
            if(conn!=null && !conn.isClosed()){
                conn.close();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void execute(Tuple tuple) {
        String word = tuple.getStringByField("word");
        if(!"QDU".equals(word)){
            count++;
            System.out.println("count=" + count + ",  word=" + word);
            if (wordList.containsKey(word)) {
                Integer tmpNum = wordList.get(word);
                tmpNum++;
                System.out.println("word=" + word + "-----count=" + tmpNum);
                wordList.put(word, tmpNum);
            } else {
                System.out.println("word=" + word + "-----count=" + 1);
                wordList.put(word, 1);
            }
            if (count%offset==0){
                for (String key : wordList.keySet()) {
                    updataData(key,wordList.get(key));
                }
            }
        }else {
            for (String key : wordList.keySet()) {
                updataData(key,wordList.get(key));
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    public void updataData(String name , Integer number){
        try {
            ps= conn.prepareStatement("select * from salenumber where sname=?");
            ps.setObject(1,name);
            rs= ps.executeQuery();
            if (rs.next()){
                ps= conn.prepareStatement("update salenumber set snumber=? where sname=?");
                ps.setObject(1,number);
                ps.setObject(2,name);
                ps.executeUpdate();
            }
            else {
                ps= conn.prepareStatement("insert into salenumber(sname,snumber) values (?,?)");
                ps.setObject(1,name);
                ps.setObject(2,number);
                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}

